package controller

import (
	"context"
	"fmt"
	"kafka/common/handlers/conf"
	"log"
	"strings"
	"time"

	"github.com/Shopify/sarama"
	"kafka/idl/proto"
)

type SendController struct {
}

// Send send
func (this SendController) Send(ctx context.Context, req *proto.PingReq) (*proto.PingRsp, error) {
	//配置发布者
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	//config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Version = sarama.V0_11_0_2
	config.Producer.Timeout = 2 * time.Second

	address := strings.Split(conf.Viper.GetString("kafka.address"), ",")

	retMsg := make([]string, 0)
	//确认返回，记得一定要写，因为本次例子我用的是同步发布者
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	//设置超时时间 这个超时时间一旦过期，新的订阅者在这个超时时间后才创建的，就不能订阅到消息了
	config.Producer.Timeout = 5 * time.Second
	//连接发布者，并创建发布者实例
	p, err := sarama.NewSyncProducer(address, config)
	if err != nil {
		log.Printf("sarama.NewSyncProducer err, message=%s \n", err)
		return nil, err
	}
	//程序退出时释放资源
	defer p.Close()
	//设置一个逻辑上的分区名
	topic := "topic001"
	//这个是发布的内容
	srcValue := "sync: this is a message. index=%d"
	//发布者循环发送0-9的消息内容
	for i := 0; i < 10; i++ {
		value := fmt.Sprintf(srcValue, i)
		//创建发布者消息体
		msg := &sarama.ProducerMessage{
			Topic: topic,
			Value: sarama.ByteEncoder(value),
		}
		//发送消息并返回消息所在的物理分区和偏移量
		partition, offset, err := p.SendMessage(msg)
		if err != nil {
			log.Printf("send message(%s) err=%s \n", value, err)
		} else {
			retMsg = append(retMsg, fmt.Sprintf(value+"发送成功，partition=%d, offset=%d \n", partition, offset))
		}
	}

	return &proto.PingRsp{
		Errno:  0,
		Errmsg: "ok",
		Data:   strings.Join(retMsg, " - "),
	}, nil
}
